/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.prediction.recset;

import aiaudio.Application;
import aiaudio.database.tables.RatingsListTable;
import aiaudio.database.tables.RatingsMatrixTable;
import java.io.IOException;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;

/**
 *
 * @author nastya
 */
public class RecommendationSetExtractorReducer extends TableReducer<ImmutableBytesWritable, ImmutableBytesWritable, ImmutableBytesWritable> {

    public static final String INITIAL_DATA_SET_NAME = "initialDataSetName";
    HTable dataSet;

    @Override
    protected void reduce(ImmutableBytesWritable key, Iterable<ImmutableBytesWritable> values, Context context) throws IOException, InterruptedException {
        for (ImmutableBytesWritable immutableBytesWritable : values) {
            addEmptyRating(key, immutableBytesWritable, context);
        }
    }

    protected void addEmptyRating(ImmutableBytesWritable key, ImmutableBytesWritable immutableBytesWritable, Context context) throws IOException, InterruptedException {
        byte[] row = Bytes.add(key.get(), immutableBytesWritable.get());
        Put put = new Put(row);
        byte[] artist = key.get();
        byte[] user = immutableBytesWritable.get();
        RatingsListTable.addArtist(put, artist);
        RatingsListTable.addUser(put, user);
        byte[] rating = existsInDataSet(artist, user);
        if (rating != null) {
            RatingsListTable.addInitialRating(put, rating);
        }
        context.write(new ImmutableBytesWritable(), put);
    }

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        super.setup(context);
        String tableName = context.getConfiguration().get(INITIAL_DATA_SET_NAME);
        dataSet = Application.database().getByName(tableName).getTable();
    }

    private byte[] existsInDataSet(byte[] artist, byte[] user) throws IOException {
        return RatingsMatrixTable.getFromDataSet(dataSet, user, artist);
    }
}
